package com.bw.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.MyKafkaUtil;
import com.bw.bean.OrderDetail;
import com.bw.bean.OrderInfo;
import com.bw.bean.OrderInfoWide1;
import com.bw.fun.MyAsyncFunction;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

public class Test4 {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env=StreamExecutionEnvironment.createLocalEnvironment();
        env.setParallelism(1);

        DataStream<String> orderInfoStream = env.addSource(MyKafkaUtil.getKafkaConsumer("dwd_order_info_topic_yk8", "test" + System.currentTimeMillis()));
        DataStream<String> orderDetailStream = env.addSource(MyKafkaUtil.getKafkaConsumer("dwd_order_detail_topic_yk8", "test" + System.currentTimeMillis()));

//        orderInfoStream.print();

        // 清洗数据
        SingleOutputStreamOperator<JSONObject> etlStream = orderInfoStream.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String s, Collector<JSONObject> collector) throws Exception {
                try {
                    if (s != null){
                        JSONObject jsonObject = JSON.parseObject(s);
                        collector.collect(jsonObject);
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });
        // 添加水位线
        SingleOutputStreamOperator<JSONObject> orderInfoWm = etlStream.assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(2)).withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
            @Override
            public long extractTimestamp(JSONObject jsonObject, long l) {
                return jsonObject.getJSONObject("data").getLong("create_time");
            }
        }));

        SingleOutputStreamOperator<JSONObject> etlStream1 = orderDetailStream.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String s, Collector<JSONObject> collector) throws Exception {
                try {
                    if (s != null){
                        JSONObject jsonObject = JSON.parseObject(s);
                        String sku_id = jsonObject.getJSONObject("data").getString("product_sku_id");
                        if (sku_id != null){
                            collector.collect(jsonObject);
                        }
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });
//        // 添加水位线
        SingleOutputStreamOperator<JSONObject> orderDetailWm = etlStream1.assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(2)).withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
            @Override
            public long extractTimestamp(JSONObject jsonObject, long l) {
                return jsonObject.getJSONObject("data").getLong("create_time");
            }
        }));
//
//        // 采用Inteval join 关联
//
        SingleOutputStreamOperator<OrderInfoWide1> process = orderDetailWm
                .keyBy(new KeySelector<JSONObject, String>() {
                    @Override
                    public String getKey(JSONObject jsonObject) throws Exception {
                        return jsonObject.getJSONObject("data").getString("order_info_id");
                    }
                })
                .intervalJoin(orderInfoWm.keyBy(new KeySelector<JSONObject, String>() {
                    @Override
                    public String getKey(JSONObject jsonObject) throws Exception {
                        return jsonObject.getJSONObject("data").getString("id");
                    }
                }))
                .between(Time.milliseconds(-5), Time.milliseconds(5))
                .process(new ProcessJoinFunction<JSONObject, JSONObject, OrderInfoWide1>() {
                    @Override
                    public void processElement(JSONObject jsonObject, JSONObject jsonObject2, ProcessJoinFunction<JSONObject, JSONObject, OrderInfoWide1>.Context context, Collector<OrderInfoWide1> collector) throws Exception {
                        String data = jsonObject2.getJSONObject("data").toJSONString();
                        OrderInfo orderInfo = JSON.parseObject(data, OrderInfo.class);


                        String data1 = jsonObject.getJSONObject("data").toJSONString();
                        OrderDetail orderDetail = JSON.parseObject(data1, OrderDetail.class);





                        collector.collect(new OrderInfoWide1(orderInfo, orderDetail));
                    }
                });

//
        SingleOutputStreamOperator<OrderInfoWide1> orderInfoWide1 = AsyncDataStream.unorderedWait(process, new MyAsyncFunction(), 1000, TimeUnit.SECONDS, 30);
        orderInfoWide1.print();
        orderInfoWide1.map(JSON::toJSONString).addSink(MyKafkaUtil.getKafkaProducer("dws_trade_orders"));
        env.execute();


    }
}
